package com.yugao.fintech.draper.message.queue.client;

import com.hazelcast.collection.IQueue;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.topic.ITopic;
import com.yugao.fintech.draper.message.queue.interceptor.HazelcastMessageInterceptor;
import com.yugao.fintech.draper.message.queue.message.AbstractBroadcastMessage;
import com.yugao.fintech.draper.message.queue.message.AbstractGroupMessage;
import com.yugao.fintech.draper.message.queue.support.broadcast.HazelcastBroadcastMessage;
import com.yugao.fintech.draper.message.queue.support.group.HazelcastGroupMessage;
import lombok.RequiredArgsConstructor;

import java.util.List;

@RequiredArgsConstructor
public class HazelcastMQTemplate extends MessageQueueTemplate<HazelcastMessageInterceptor> {

    private final HazelcastInstance hazelcastInstance;

    private final List<HazelcastMessageInterceptor> interceptors;

    /**
     * 发送分组消息
     *
     * @param message 消费者组消息
     * @param <T>     消费者组消息
     */
    protected <T extends AbstractGroupMessage> void doSend(T message) {
        String groupName = message.getGroup();
        IQueue<T> queue = hazelcastInstance.getQueue(groupName);
        try {
            sendMessageBefore(message);
            queue.put(message);
            sendMessageBefore(message);
        } catch (InterruptedException e) {
        }
    }

    /**
     * 发送广播消息
     *
     * @param message 广播消息
     * @param <T>     广播消息
     */
    protected <T extends AbstractBroadcastMessage> void doSend(T message) {
        String topicName = message.getTopic();
        ITopic<T> topic = hazelcastInstance.getTopic(topicName);
        topic.publish(message);
    }

    private void sendMessageBefore(HazelcastGroupMessage message) {
        interceptors.forEach(interceptor -> interceptor.sendMessageBefore(message));
    }

    private void sendMessageAfter(HazelcastBroadcastMessage message) {
        for (int i = interceptors.size() - 1; i >= 0; i--) {
            interceptors.get(i).sendMessageAfter(message);
        }
    }
}
